package com.example.kafka;

import com.alibaba.dubbo.common.utils.ConcurrentHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by dong peng on 2019/11/16.
 */

public class EventProcess {

    private ConcurrentHashMap<String, String> hasTopicInfo = new ConcurrentHashMap<>();

    private Logger logger = LoggerFactory.getLogger(EventProcess.class);

    public void receiveConfigUpdate(TopicInfo topicInfo) {
        if(topicInfo == null) {
            logger.info("主题配置不能为空");
            return;
        }

        String topic = topicInfo.getTopic();
        String ips = topicInfo.getIps();
        if(hasTopicInfo.containsKey(topic)) {
            if(!hasTopicInfo.containsValue(ips)) {
                if(ips.contains("127.0.0.1")) {
                    System.out.println("修改==========");
                    KafkaTopicEvent.getInstance().UpdateSubscribeEvent(topicInfo);
                    hasTopicInfo.put(topic, ips);
                }else {
                    System.out.println("移除==========");
                    KafkaTopicEvent.getInstance().DelSubscribeEvent(topicInfo);
                    hasTopicInfo.remove(topic);
                }


            }
        }else {
            System.out.println("新增==========");
            KafkaTopicEvent.getInstance().AddSubscribeEvent(topicInfo);
            hasTopicInfo.put(topic, ips);
        }

    }
}
